package com.wuba.rxjavademo.lyz.day0416;

import java.util.concurrent.TimeUnit;

import rx.Notification;
import rx.Observable;
import rx.schedulers.Schedulers;
import rx.schedulers.TimeInterval;
import rx.schedulers.Timestamped;

/**
 * Created by liyaxing on 2016/4/19.
 */
public class Demo06 {

    public static void main(String[] ss){

        Demo06 demo = new Demo06() ;
        Observable<Integer> observable = demo.timeoutObservable() ;
        observable.subscribe(integer ->{
            print("value : " + integer) ;
        }) ;


        try {
            Thread.sleep(60000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

//    private Observable usingObservable(){
//        return Observable.using(()->Observable.just(1,2,3,4,5),integerObservable -> )
//    }

    private Observable<Integer> timeoutObservable(){
        return createObservable().timeout(1000,TimeUnit.MILLISECONDS) ;
    }

    private Observable<Timestamped<Integer>> timeStampedObservable(){
        return createObservable().timestamp() ;
    }

    private Observable<TimeInterval<Integer>> timeIntervalObservable(){
        return createObservable().timeInterval() ;
    }

    private Observable<Integer> threadObservable(){
        return createObservable().subscribeOn(Schedulers.immediate()).observeOn(Schedulers.newThread()) ;
    }

    private Observable<Integer> deMaterializeObservable(){
        return materializeObservable().dematerialize() ;
    }

    private Observable<Notification<Integer>> materializeObservable(){
        return createObservable().materialize() ;
    }

    private Observable<Integer> doObservable(){
        return createObservable().doOnEach(notification -> {
            print(notification.getKind() + "::" + notification.getThrowable() + "::" + notification.getValue());
        }) ;
    }

    private Observable<Integer> delaySubscriptionObservable(){
        return createObservable().delaySubscription(5000,TimeUnit.MILLISECONDS) ;
    }

    private Observable<Integer> delayObservable(){
        return createObservable().delay(5000, TimeUnit.MILLISECONDS) ;
    }

    private Observable<Integer> createObservable(){
        return Observable.<Integer>create(subscriber -> {

            for (int i = 0; i < 20; i++) {
                if(!subscriber.isUnsubscribed()){
                    subscriber.onNext(i);
                }
                try {
                    if(i == 10){
                        Thread.sleep(600);
                    }
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            subscriber.onCompleted();

        }).subscribeOn(Schedulers.newThread()) ;
    }

    private static void print(Object obj){
        System.out.println(obj) ;
    }

}
